package com.neusoft.bd.myflink.mytable.step2;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class QueryWithTableAPI {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(settings);
//        创建原表
        tableEnvironment.executeSql( " CREATE TABLE KafkaTable (\n" +
                "  `sid` STRING,\n" +
                "  `call_in` STRING,\n" +
                "  `call_out` STRING,\n" +
                "  `call_type` STRING,\n" +
                "  `call_time` BIGINT,\n" +
                "  `duration` BIGINT \n" +

                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'test',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'csv'\n" +
                ") ");

        Table table = tableEnvironment.from("   KafkaTable ");
        Table filter = table.filter($("call_type").isEqual("success").and($("duration").isGreaterOrEqual(10)));

        tableEnvironment.sqlQuery(   "    select  sid, sum('duration')    from   " +  filter  +  "   group by sid      " ).execute().print();

    }



}
